
#undef LOGGER
#define NDEBUG

#include <stdint.h>
#include <ext/hash_map>
#include <ext/hash_set>
#include <utility>

#include <appframe/circular_array_queue.hpp>
#include <xsocket/imessageprocessor.hpp>
#include <xsocket/connectioninfo.hpp>
#include <xsocket/active_reactor.hpp>

#include <xbase_datagram_factory.hpp>
#include <xbase_utility.hpp>
#include <stringable.hpp>

class ClientMessageProcessor : public IMessageProcessor
			     , public stringable
{

public:
	// typedefs
	typedef __gnu_cxx::hash_map<uint32_t, ConnectionInfo* > ConnectionMap;
	typedef __gnu_cxx::hash_set<ConnectionInfo* > ConnectionSet;

	typedef std::pair<uint32_t, ByteBuffer*> MessageType;
	typedef CircularArrayQueue<MessageType> MessageQueue;
	
public:

	ClientMessageProcessor ();

	virtual ~ClientMessageProcessor ();

	virtual void accept (ConnectionInfo &conn);
	virtual void remove (ConnectionInfo &conn);

	virtual void process_input (ConnectionInfo &conn, ByteBuffer &bb);
	virtual void process_output ();

	virtual int frame (ByteBuffer &bb);

	// for debug
	std::string to_string () const {
		std::stringstream out;
		out << "{handlerNumber:" << _handlerNumber
		    << "; connections_size:" << _connections.size ()
		    << "; connections_by_id_size:" << _connections_by_id.size ()
		    << "}";
		
		return out.str();
	}

private:
	// TODO: 用线程池来实现多个Handler
	
	// 客户端请求的处理线程组
	unsigned int _handlerNumber;
	void **_handlers;
	unsigned int m_cur_handler;

	ConnectionSet _connections;
	ConnectionMap _connections_by_id;
	// unsigned int _max_connetions;

	MessageQueue m_queue;

	// for debug & profile
	unsigned int _input_count;
	unsigned int _sent_input_count;
	unsigned int _output_count;
	unsigned int _sent_output_count;
};

// -*-mode:c++; coding:utf-8-*-

#include <exception>
#include <utility>
#include <algorithm>
using namespace std;

#include <boost/foreach.hpp>
#include <boost/scoped_ptr.hpp>

#include <xsocket/bytebuffer.hpp>

#include <xbase_exception.hpp>
#include <xbase_utility.hpp>

ClientMessageProcessor::ClientMessageProcessor ()
	: _handlerNumber(0)
	, _handlers(NULL)
	, m_cur_handler(0)
	, m_queue(10240)
	, _input_count(0)
	, _output_count(0)
{
}


ClientMessageProcessor::~ClientMessageProcessor () {
	// first save connection in list
	std::vector<ConnectionInfo*> _list;
	BOOST_FOREACH(ConnectionInfo *_p, _connections)
	{
		_list.push_back(_p);
	}
	// then close all alive connection
	BOOST_FOREACH(ConnectionInfo *_p, _list)
	{
		_p->close();
	}
}

void ClientMessageProcessor::process_input (ConnectionInfo &conn, ByteBuffer &bb) {
	// for debug
	++ _input_count;

	int conn_id = conn.get_id ();

	// 拷贝到新的bytebuffer
	//boost::shared_ptr<ByteBuffer> buffer(new ByteBuffer(bb.remaining()));
	std::auto_ptr<ByteBuffer> buffer(new ByteBuffer(std::max((int)MAX_PACKET_SIZE, bb.remaining())));
	buffer->put(bb);
	buffer->flip();
	// ByteBuffer::arraycopy (bb.array (), bb.position (),
	// 		       buffer->array (), buffer->position (),
	// 		       bb.remaining ());
	// buffer->limit(bb.remaining ());
	
	MessageType msg(conn_id, buffer.get());
	ScanCloseReq req;
	XBaseDatagramFactory::decode(*msg.second, req);

	MessageType resp_msg = msg;
	ScanCloseResp resp;
	resp_msg.second->clear();
	XBaseDatagramFactory::encode(*msg.second, resp);

	// 选择一个处理线程进行处理
	// TODO: 用更好的选择策略，当前的策略是同一个Socket的数据被放到相同的处理线程
	//_handlers [conn_id % _handlerNumber]->sendMessage (msg);
	//_handlers[m_cur_handler]->sendMessage (msg);
	m_queue.push(resp_msg);
	buffer.release();

	++m_cur_handler;
	if(m_cur_handler >= _handlerNumber) {
		m_cur_handler = 0;
	}

	// for debug
	++ _sent_input_count;
}

void ClientMessageProcessor::process_output () {
	// for load info

	//MessageType msg;
	unsigned int count_per_handler = 0;

	for (int j=0; j<128; ++j) {
		if(m_queue.empty())
			break;
		
		MessageType msg = m_queue.front();
		m_queue.pop();
		
		if ((msg.first == 0) && (msg.second == 0)) // no message
			break;
		
		// for debug
		// 记录成功取出及总的的响应包数，用于统计。
		++ count_per_handler;
		++ _output_count;

		boost::scoped_ptr<ByteBuffer> buffer(msg.second);
		if(buffer)
			buffer->flip();
		
		ConnectionMap::iterator iter = _connections_by_id.find (msg.first);
		
		// 没有该连接，可能是处理期间已经关闭或
		// 未知的连接，直接将其丢弃。
		if (iter == _connections_by_id.end()) {
			ostringstream stream;
			if (buffer)
				stream << CommandName(XBaseDatagramFactory::getType(*buffer));
			else
				stream << "NULL";
			
			continue;
		}
		
		// 通常情况下，连接断开就会从列表中删除
		// 的，所以这里取到的一定是存活的连接，
		// 但以防未知的错误发生，这里进行了一次
		// 检查
		
		// 若是已关闭连接，则不应该出现在该列表
		// 中，检测出来后将其删除，并将相关的响
		// 应包丢弃
		if(! iter->second->is_open()) {
			ostringstream stream;
			if (buffer)
				stream << CommandName(XBaseDatagramFactory::getType(*buffer));
			else
				stream << "NULL";
			
			this->remove(*iter->second); // must first
			iter->second->close(); // must second
			continue;
		}
		
		// 空的数据指针，则将连接关闭
		if (! buffer) {

			// 前面已经保证该连接存在并没有
			// 关闭，因此只调用close即可
			iter->second->close();
			continue;
		}
		
		// 若是一个空的数据包，也将连接关闭
		if (buffer->remaining () == 0) {
			iter->second->close();
			continue;
		}
				
		// 若没准备好写，则可能是上次的没有发送
		// 完，或连接出问题了，尝试将其flush一
		// 下再作处理
		if(! iter->second->ready_to_write()) {
			iter->second->flush ();
		}
		
		// 若仍然不能写，很可能是出问题了，也将
		// 其关闭。
		if(! iter->second->ready_to_write()) {
			iter->second->close ();
			continue;
		}
		
		// 到现在才排除了各种可能的问题和错误，开始发送包
		int ret = iter->second->write (*buffer);
				
		// 若没发送成功，则将连接关闭
		if(ret < 0) {

			// 关闭操作会调用自身的remove函
			// 数将其删除
			iter->second->close ();
			continue;
		}

		// 到此才发送成功，设置统计用的计数器
		++ _sent_output_count;
	}	
}

void ClientMessageProcessor::accept (ConnectionInfo &conn) {
	// DONE: 这里不需要考虑插入失败吗？虽然基本上不会有两个相同的
	// Key，除非是同一个对象。而这时候它已经在里面了，就不用再插入
	// 了吧。
	_connections.insert (&conn);
	_connections_by_id.insert (ConnectionMap::value_type(conn.get_id(), &conn));

	//cout << "accepted " << conn << endl;
}

void ClientMessageProcessor::remove (ConnectionInfo &conn) {
	_connections.erase (&conn);
	_connections_by_id.erase (conn.get_id());
	
	//cout<< "removed " << conn << endl;
}

int ClientMessageProcessor::frame (ByteBuffer &bb) {
	return XBaseDatagramFactory::frame (bb);
}

int main(int argc, char **argv)
{
	ClientMessageProcessor processor;
	ActiveReactor reactor;
	reactor.init();
	const int ret = reactor.add_acceptor("0.0.0.0", 20083, processor);
	if(ret < 0)
	{
		return -1;
	}

	while(true)
	{
		reactor.poll();

		// wasting time
		int n=9;
		time_t time = cur_micro_seconds();
		while(n--)
		{
			time += cur_micro_seconds();
		}
	}

	return 0;
}
